package com.smsc.headend.task.engine.dao.impl;

import cn.hutool.core.date.SystemClock;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.task.engine.dao.RdMeterReadsLpDAO;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.List;

@Repository
@Slf4j
public class RdMeterReadsLpDAOImpl implements RdMeterReadsLpDAO {

    @Autowired
    DataSource dataSource;


    @Override
    public Object batchSaveRdMeterReadsLp(List<RdMeterReadsLp> rdMeterReadsLps, String dbName, String yearMonth) {
        long start = SystemClock.now();
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("MERGE INTO RD_METER_READS_LP%s A")
                .append("    USING (SELECT ? AS METER_ID, ? AS SRC, ? AS TV, ? AS DATA_ITEM_ID, ? AS LP_DATA_ITEM_ID, ? AS VALUE, ? AS VALIDITY,? AS TV_UPDATE FROM DUAL) B ON ( A.LP_DATA_ITEM_ID=B.LP_DATA_ITEM_ID AND A.DATA_ITEM_ID=B.DATA_ITEM_ID AND A.TV=B.TV AND A.METER_ID =B.METER_ID)")
                .append("    WHEN MATCHED THEN UPDATE SET A.VALUE=B.VALUE, A.SRC=B.SRC, A.VALIDITY=B.VALIDITY, A.TV_UPDATE=B.TV_UPDATE ")
                .append("    WHEN NOT MATCHED THEN INSERT(METER_ID, SRC, TV, DATA_ITEM_ID, LP_DATA_ITEM_ID, VALUE, VALIDITY, TV_UPDATE)")
                .append("    VALUES(B.METER_ID, B.SRC, B.TV, B.DATA_ITEM_ID, B.LP_DATA_ITEM_ID, B.VALUE, B.VALIDITY, B.TV_UPDATE)");

        String sql = stringBuilder.toString();
        String postfix = "_" + yearMonth;

        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dbName);
        sql = String.format(sql, postfix);
        log.debug(sql);
        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (RdMeterReadsLp batchAddMeterRead : rdMeterReadsLps) {
                ps.setLong(1, batchAddMeterRead.getMeterId());
                ps.setInt(2, batchAddMeterRead.getSrc());
                ps.setLong(3, batchAddMeterRead.getTv());
                ps.setLong(4, batchAddMeterRead.getDataItemId());
                ps.setLong(5, batchAddMeterRead.getLpDataItemId());
                ps.setString(6, batchAddMeterRead.getValue());
                ps.setInt(7, batchAddMeterRead.getValidity());
                ps.setLong(8, batchAddMeterRead.getTvUpdate());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            log.debug("save lp records size:{},cost: {}ms", rdMeterReadsLps.size(), SystemClock.now() - start);
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }
        return true;
    }

    @Override
    public Object batchSaveRdMeterReadsLpForPgsql(List<RdMeterReadsLp> rdMeterReadsLps,String dbName, String yearMonth) {
//        log.info("start-end:{}->{}", start, after);
        long start = SystemClock.now();
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("with upsert as ( ");
        stringBuilder.append("    update RD_METER_READS_LP%s A");
        stringBuilder.append("        set VALUE=B.VALUE, SRC=B.SRC, VALIDITY=B.VALIDITY, TV_UPDATE=B.TV_UPDATE");
        stringBuilder.append("        from (SELECT ? AS METER_ID, ? AS SRC, ? AS TV, ? AS DATA_ITEM_ID, ? AS LP_DATA_ITEM_ID, ? AS VALUE, ? AS VALIDITY,? AS TV_UPDATE ) B");
        stringBuilder.append("        where A.LP_DATA_ITEM_ID=B.LP_DATA_ITEM_ID AND A.DATA_ITEM_ID=B.DATA_ITEM_ID AND A.TV=B.TV AND A.METER_ID =B.METER_ID");
        stringBuilder.append("        returning A.LP_DATA_ITEM_ID,A.DATA_ITEM_ID,A.TV,A.METER_ID)");
        stringBuilder.append(" insert into RD_METER_READS_LP%s ");
        stringBuilder.append(" select *");
        stringBuilder.append(" from (SELECT ? AS METER_ID, ? AS SRC, ? AS TV, ? AS DATA_ITEM_ID, ? AS LP_DATA_ITEM_ID, ? AS VALUE, ? AS VALIDITY,? AS TV_UPDATE) t");
        stringBuilder.append(" where not exists(");
        stringBuilder.append("        select 1");
        stringBuilder.append("        from upsert B");
        stringBuilder.append("        where t.LP_DATA_ITEM_ID = B.LP_DATA_ITEM_ID and t.DATA_ITEM_ID = B.DATA_ITEM_ID and t.TV = B.TV and t.METER_ID = B.METER_ID )");

        String sql = stringBuilder.toString();

        String postfix = "_" + yearMonth;
//
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dbName);

        sql = String.format(sql, postfix,postfix);
        log.debug(sql);
        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (RdMeterReadsLp batchAddMeterRead : rdMeterReadsLps) {
                ps.setLong(1, batchAddMeterRead.getMeterId());
                ps.setInt(2, batchAddMeterRead.getSrc());
                ps.setLong(3, batchAddMeterRead.getTv());
                ps.setLong(4, batchAddMeterRead.getDataItemId());
                ps.setLong(5, batchAddMeterRead.getLpDataItemId());
                ps.setString(6, batchAddMeterRead.getValue());
                ps.setInt(7, batchAddMeterRead.getValidity());
                if(batchAddMeterRead.getTvUpdate() == null) {
                    ps.setNull(8, Types.BIGINT);
                } else {
                    ps.setLong(8, batchAddMeterRead.getTvUpdate());
                }
                ps.setLong(9, batchAddMeterRead.getMeterId());
                ps.setInt(10, batchAddMeterRead.getSrc());
                ps.setLong(11, batchAddMeterRead.getTv());
                ps.setLong(12, batchAddMeterRead.getDataItemId());
                ps.setLong(13, batchAddMeterRead.getLpDataItemId());
                ps.setString(14, batchAddMeterRead.getValue());
                ps.setInt(15, batchAddMeterRead.getValidity());
                if(batchAddMeterRead.getTvUpdate() == null) {
                    ps.setNull(16, Types.BIGINT);
                } else {
                    ps.setLong(16, batchAddMeterRead.getTvUpdate());
                }
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            log.debug("save lp records size:{},cost: {}ms", rdMeterReadsLps.size(), SystemClock.now() - start);
            ps.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }

        return true;
    }



}
